package com.cyj.kafka.broadcast;

import com.cyj.kafka.subscribe.SubscribeConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/**
 * @ClassName BroadcastConsumer
 * @Description TODO
 * @Author chixiaowai
 * @Date 2022/10/23 22:58
 * @Version 1.0
 */

public class BroadcastConsumer implements Runnable{
    private KafkaConsumer<String, String> kafkaConsumer;
    private ConsumerRecords<String, String> consumerRecords;
    private static String groupId = "group";

    public BroadcastConsumer(List<String> topics, String groupId){
        //配置
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop101:9092,hadoop102:9092,hadoop103:9092");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        //创建一个消费者
        kafkaConsumer = new KafkaConsumer<String, String>(properties);
        //订阅主题
        kafkaConsumer.subscribe(topics);
    }

    @Override
    public void run() {
        try {
            int count = 1;
            //消费数据
            while (true){
                //1秒钟拉取一次
                consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
                if (consumerRecords != null && consumerRecords.count() > 0){
                    for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                        System.out.println("topic:" + consumerRecord.topic() + "\t" + "offset:" + consumerRecord.offset() + "\t" +
                                "key:" + consumerRecord.key() + "\t" + "value:" + consumerRecord.value());
                        if (count % 100 == 0){
                            break;
                        }
                    }
                }else {
                    Thread.sleep(1000);
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            kafkaConsumer.close();
        }
    }

    public static void main(String[] args) {
        ArrayList<String> topics = new ArrayList<>();
        topics.add("first");
        topics.add("second");
        for (int i = 1; i <= 5; i++) {
            SubscribeConsumer subscribeConsumer = new SubscribeConsumer(topics, (groupId + i));
            Thread thread = new Thread(subscribeConsumer);
            thread.start();
        }
    }
}
